package HandshakeJoin;

import java.util.List;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import com.google.common.collect.Lists;
import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnel;
import com.google.common.hash.PrimitiveSink;
import com.google.common.base.Joiner;
import org.apache.s4.base.Event;

public class Snapshot extends Event
{
    transient private static final double bloomFPP = 0.02;

    private String joinId;
    private long timestamp;
    private List<ArrayList<Integer>> records;
    private BloomFilter<ArrayList<Integer>> bloom;

    Snapshot(final String joinId, List<ArrayList<Integer>> records) {
        this.joinId = joinId;
        this.records = records;
        timestamp = Calendar.getInstance().getTimeInMillis();

        // constructBloomFilter();
    }

    private void constructBloomFilter() {
        bloom = BloomFilter.create(RecordFunnel.INSTANCE, records.size(),
                bloomFPP);

        for (ArrayList<Integer> record : records) {
            bloom.put(record);
        }
    }

    public String getJoinId() {
        return joinId;
    }

    public long getTimeStamp() {
        return timestamp;
    }

    public Snapshot setTimestamp(final long timestamp) {
        this.timestamp = timestamp;

        return this;
    }

    public List<ArrayList<Integer>> getRecords() {
        return records;
    }

    public int getRecordsCount() {
        return records.size();
    }

    public int getRecordColumnsCount() {
        return records.get(0).size();
    }

    public BloomFilter<ArrayList<Integer>> getBloomFilter() {
        return bloom;
    }

    public String toString() {
        StringBuilder sb = new StringBuilder();
        // Joiner joinerLine = Joiner.on("\n").skipNulls();
        Joiner joinerComma = Joiner.on(", ").skipNulls();

        DateFormat dateFormat = new SimpleDateFormat("HH:mm:ss");
        String timestampStr = dateFormat.format(new Date(timestamp));
        sb.append("### Join ID: " + joinId + " @ " + timestampStr + "\n");
        // sb.append(joinerLine.join(records));
        sb.append(joinerComma.join(records));
        sb.append("\n--------------------------------\n");

        return sb.toString();
    }

    private enum RecordFunnel implements Funnel<ArrayList<Integer>> {
        INSTANCE;
        public void funnel(ArrayList<Integer> record, PrimitiveSink into) {
            into.putInt(record.get(0));
        }
    }
}